package com.heima.example.chat.client;

import com.heima.example.chat.client.handler.RpcResponseMessageHandler;
import com.heima.example.chat.message.RpcRequestMessage;
import com.heima.example.chat.message.RpcResponseMessage;
import com.heima.example.chat.protocol.MessageCodecSharable;
import com.heima.example.chat.protocol.ProcotolFrameDecoder;
import com.heima.example.chat.protocol.SequenceIdGenerator;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.util.concurrent.DefaultPromise;
import io.netty.util.concurrent.Promise;
import lombok.extern.slf4j.Slf4j;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;

/**
 * 创建出一个 Channel 对象，通过该Channel对象来发送RPC请求
 *
 * @author ZhengDp
 * @Date 2023/7/7 11:44
 */
@Slf4j
public class RpcClientManager {

    private static volatile Channel channel = null;
    private static final Object LOCK = new Object();

    /**
     * 获取代理类
     */
    public static <T> T getProxyService(Class<T> serviceClass) {
        Object result = Proxy.newProxyInstance(serviceClass.getClassLoader(), serviceClass.getInterfaces(), new InvocationHandler() {
            @Override
            public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
                // 将方法调用转换为 消息对象
                int sequenceId = SequenceIdGenerator.nextId();
                RpcRequestMessage msg = new RpcRequestMessage(
                        sequenceId,
                        serviceClass.getName(),
                        method.getName(),
                        method.getReturnType(),
                        method.getParameterTypes(),
                        args
                );
                // 发送消息对象
                getChannel().writeAndFlush(msg);
                // 如何接受响应？ ---> 主线程发起调用，NioEvent线程中处理响应，两个线程之间的通信 ---> Promise对象来进行通信
                //                                                   指定 promise 对象异步接收结果线程
                // promise.addListener(future ->{}); --> 异步接收线程
                DefaultPromise<Object> promise = new DefaultPromise<>(getChannel().eventLoop());
                RpcResponseMessageHandler.PROMISES.put(sequenceId,promise);
                // 等待 Promise结果
                promise.await();
                if(promise.isSuccess()) {
                    return promise.get();
                } else {
                    Throwable cause = promise.cause();
                    throw new RuntimeException(cause);
                }
            }
        });
        return (T)result;
    }


    public static Channel getChannel() {
        if (channel != null) {
            return channel;
        }
        synchronized (LOCK) {
            if (channel != null) {
                return channel;
            }
            initChannel();
            return channel;

        }
    }


    private static void initChannel() {
        NioEventLoopGroup group = new NioEventLoopGroup();
        LoggingHandler LOGGING_HANDLER = new LoggingHandler(LogLevel.DEBUG);
        MessageCodecSharable MESSAGE_CODEC = new MessageCodecSharable();
        RpcResponseMessageHandler RPC_HANDLER = new RpcResponseMessageHandler();
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.channel(NioSocketChannel.class);
        bootstrap.group(group);
        bootstrap.handler(new ChannelInitializer<SocketChannel>() {
            @Override
            protected void initChannel(SocketChannel ch) throws Exception {
                ch.pipeline().addLast(new ProcotolFrameDecoder());
                ch.pipeline().addLast(LOGGING_HANDLER);
                ch.pipeline().addLast(MESSAGE_CODEC);
                ch.pipeline().addLast(RPC_HANDLER);
            }
        });
        try {
            // 客户端建立连接后，发送一个请求消息
            channel = bootstrap.connect("localhost", 7777).sync().channel();
            channel.closeFuture().addListener(future -> {
                group.shutdownGracefully();
            });
        } catch (Exception e) {
            log.error("client error", e);
        }
    }
}
